package com.blastking.sso.core.util;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.JedisShardInfo;
import redis.clients.jedis.MultiKeyPipelineBase;
import redis.clients.jedis.Pipeline;
import redis.clients.jedis.ShardedJedis;
import redis.clients.jedis.ShardedJedisPipeline;
import redis.clients.jedis.ShardedJedisPool;

import java.io.*;
import java.util.Collection;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * Redis client base on jedis 根据继承类的不同,
 * jedis实例方式不用:JedisSimpleFactry/JedisPoolFactry/ShardedJedisPoolFactry
 * 
 * @author wxb 2015-7-10 18:34:07
 *
 *         # for redis (sharded.jedis.address=host01:port,host02:port)
 *         sharded.jedis.address=127.0.0.1:6379,127.0.0.1:6379,127.0.0.1:6379
 *
 */
public class JedisUtilImpl implements JedisUtil {
	private static Logger logger = LoggerFactory.getLogger(JedisUtilImpl.class);

	private static final int DEFAULT_EXPIRE_TIME = 7200; // 默认过期时间,单位/秒, 60*60*2=2H, 两小时
	private String address;

	private int db = -1;

	public void init(String address) {
		this.address = address;
	}

	public void init(String address, int db) {
		this.address = address;
		this.db = db;
	}

	// ------------------------ ShardedJedisPool ------------------------
	/**
	 * 方式01: Redis单节点 + Jedis单例 : Redis单节点压力过重, Jedis单例存在并发瓶颈 》》不可用于线上 new
	 * Jedis("127.0.0.1", 6379).get("cache_key"); 方式02: Redis单节点 + JedisPool单节点连接池
	 * 》》 Redis单节点压力过重，负载和容灾比较差 new JedisPool(new JedisPoolConfig(), "127.0.0.1",
	 * 6379, 10000).getResource().get("cache_key"); 方式03:
	 * Redis集群(通过client端集群,一致性哈希方式实现) + Jedis多节点连接池 》》Redis集群,负载和容灾较好,
	 * ShardedJedisPool一致性哈希分片,读写均匀，动态扩充 new ShardedJedisPool(new JedisPoolConfig(),
	 * new LinkedList<JedisShardInfo>());
	 */

	private ShardedJedisPool shardedJedisPool;
	private ReentrantLock INSTANCE_INIT_LOCL = new ReentrantLock(false);

	private static JedisPoolConfig createPoolConfig() {
		JedisPoolConfig config = new JedisPoolConfig();
		config.setMaxTotal(200); // 最大连接数, 默认8个
		config.setMaxIdle(50); // 最大空闲连接数, 默认8个
		config.setMinIdle(8); // 设置最小空闲数
		config.setMaxWaitMillis(10000); // 获取连接时的最大等待毫秒数(如果设置为阻塞时BlockWhenExhausted),如果超时就抛异常, 小于零:阻塞不确定的时间, 默认-1
		config.setTestOnBorrow(true); // 在获取连接的时候检查有效性, 默认false
		config.setTestOnReturn(true); // 调用returnObject方法时，是否进行有效检查
		config.setTestWhileIdle(true); // Idle时进行连接扫描
		config.setTimeBetweenEvictionRunsMillis(30000); // 表示idle object evitor两次扫描之间要sleep的毫秒数
		config.setNumTestsPerEvictionRun(10); // 表示idle object evitor每次扫描的最多的对象数
		config.setMinEvictableIdleTimeMillis(60000); // 表示一个对象至少停留在idle状态的最短时间，然后才能被idle object
														// evitor扫描并驱逐；这一项只有在timeBetweenEvictionRunsMillis大于0时才有意义

		return config;
	}

	private static List<JedisShardInfo> createJedisShards(String address) {
		// JedisShardInfo List
		List<JedisShardInfo> jedisShardInfos = new LinkedList<JedisShardInfo>();

		String[] addressArr = address.split(",");
		for (int i = 0; i < addressArr.length; i++) {
			String[] addressInfo = addressArr[i].split(":");
			String host = addressInfo[0];
			int port = Integer.valueOf(addressInfo[1]);
			JedisShardInfo jedisShardInfo = new JedisShardInfo(host, port, 10000);
			jedisShardInfos.add(jedisShardInfo);
		}
		return jedisShardInfos;
	}

	private static ShardedJedisPool createJedisPool(String address) {

		// JedisShardInfo List
		List<JedisShardInfo> jedisShardInfos = createJedisShards(address);
		// JedisPoolConfig
		JedisPoolConfig config = createPoolConfig();
		ShardedJedisPool pool = new ShardedJedisPool(config, jedisShardInfos);
		logger.info(">>>>>>>>>>> blastking-sso, JedisUtil.ShardedJedisPool init success.");
		return pool;
	}

	private void createJedisWithLock(String address) {
		try {
			if (INSTANCE_INIT_LOCL.tryLock(2, TimeUnit.SECONDS)) {
				try {

					if (shardedJedisPool == null) {
						shardedJedisPool = createJedisPool(address);
					}
					return;

				} finally {
					INSTANCE_INIT_LOCL.unlock();
				}
			}

		} catch (InterruptedException e) {
			logger.error(e.getMessage(), e);
		}
	}

	/**
	 * 获取ShardedJedis实例
	 * 
	 * @return
	 */
	private ShardedJedis getInstance() {
		if (null == shardedJedisPool) {
			createJedisWithLock(address);
		}

		if (null == shardedJedisPool) {
			throw new NullPointerException(">>>>>>>>>>> blastking-sso, JedisUtil.ShardedJedisPool is null.");
		}

		ShardedJedis shardedJedis = shardedJedisPool.getResource();

		if (db != -1) {
			Collection<Jedis> collection = shardedJedis.getAllShards();
			Iterator<Jedis> jedis = collection.iterator();
			while (jedis.hasNext()) {
				jedis.next().select(db);		
			}
		}
		return shardedJedis;
	}

	// ------------------------ seblastkinglize and unserialize
	// ------------------------
	/**
	 * 将对象-->byte[] (由于jedis中不支持直接存储object所以转换成byte[]存入)
	 *
	 * @param object
	 * @return
	 */
	private static byte[] serialize(Object object) {
		ObjectOutputStream oos = null;
		ByteArrayOutputStream baos = null;
		try {
			// 序列化
			baos = new ByteArrayOutputStream();
			oos = new ObjectOutputStream(baos);
			oos.writeObject(object);
			byte[] bytes = baos.toByteArray();
			return bytes;
		} catch (Exception e) {
			logger.error("{}", e);
		} finally {
			try {
				oos.close();
				baos.close();
			} catch (IOException e) {
				logger.error("{}", e);
			}
		}
		return null;
	}

	/**
	 * 将byte[] -->Object
	 *
	 * @param bytes
	 * @return
	 */
	private static Object unserialize(byte[] bytes) {
		ByteArrayInputStream bais = null;
		try {
			// 反序列化
			bais = new ByteArrayInputStream(bytes);
			ObjectInputStream ois = new ObjectInputStream(bais);
			return ois.readObject();
		} catch (Exception e) {
			logger.error("{}", e);
		} finally {
			try {
				bais.close();
			} catch (IOException e) {
				logger.error("{}", e);
			}
		}
		return null;
	}

	// ------------------------ jedis util ------------------------
	/**
	 * 存储简单的字符串或者是Object 因为jedis没有分装直接存储Object的方法，所以在存储对象需斟酌下
	 * 存储对象的字段是不是非常多而且是不是每个字段都用到，如果是的话那建议直接存储对象，
	 * 否则建议用集合的方式存储，因为redis可以针对集合进行日常的操作很方便而且还可以节省空间
	 */

	/**
	 * Set String
	 * 
	 * @param key
	 * @param value
	 * @param seconds
	 *            存活时间,单位/秒
	 * @return
	 */
	public String setStringValue(String key, String value, int seconds) {
		return hasRetFun((client) -> {
			return client.setex(key, seconds, value);
		}, null);
	}

	/**
	 * Set String (默认存活时间, 2H)
	 * 
	 * @param key
	 * @param value
	 * @return
	 */
	public String setStringValue(String key, String value) {
		return setStringValue(key, value, DEFAULT_EXPIRE_TIME);
	}

	/**
	 * Set Object
	 *
	 * @param key
	 * @param obj
	 * @param seconds
	 *            存活时间,单位/秒
	 */
	public String setObjectValue(String key, Object obj, int seconds) {
		return hasRetFun((client) -> {
			return client.setex(key.getBytes(), seconds, serialize(obj));
		}, null);
	}

	/**
	 * Set Object (默认存活时间, 2H)
	 * 
	 * @param key
	 * @param obj
	 * @return
	 */
	public String setObjectValue(String key, Object obj) {
		return setObjectValue(key, obj, DEFAULT_EXPIRE_TIME);
	}

	/**
	 * Get String
	 * 
	 * @param key
	 * @return
	 */
	public String getStringValue(String key) {
		return hasRetFun((client) -> {
			return client.get(key);
		}, null);
	}

	/**
	 * Get Object
	 * 
	 * @param key
	 * @return
	 */
	public Object getObjectValue(String key) {
		return hasRetFun((client) -> {
			Object obj = null;
			byte[] bytes = client.get(key.getBytes());
			if (bytes != null && bytes.length > 0) {
				obj = unserialize(bytes);
			}
			return obj;
		}, null);
	}

	/**
	 * Delete
	 * 
	 * @param key
	 * @return Integer reply, specifically: an integer greater than 0 if one or more
	 *         keys were removed 0 if none of the specified key existed
	 */
	public Long del(String key) {
		return hasRetFun((client) -> {
			return client.del(key);
		}, null);
	}

	/**
	 * incrBy value值加i
	 * 
	 * @param key
	 * @param i
	 * @return new value after incr
	 */
	public Long incrBy(String key, int i) {
		return hasRetFun((client) -> {
			return client.incrBy(key, i);
		}, null);
	}

	/**
	 * exists
	 * 
	 * @param key
	 * @return Boolean reply, true if the key exists, otherwise false
	 */
	public boolean exists(String key) {
		return hasRetFun((client) -> {
			return client.exists(key);
		}, null);
	}

	/**
	 * expire 重置存活时间
	 * 
	 * @param key
	 * @param seconds
	 *            存活时间,单位/秒
	 * @return Integer reply, specifically: 1: the timeout was set. 0: the timeout
	 *         was not set since the key already has an associated timeout (versions
	 *         lt 2.1.3), or the key does not exist.
	 */
	public long expire(String key, int seconds) {
		return hasRetFun((client) -> {
			return client.expire(key, seconds);
		}, null);
	}

	/**
	 * expireAt 设置存活截止时间
	 * 
	 * @param key
	 * @param unixTime
	 *            存活截止时间戳
	 * @return
	 */
	public long expireAt(String key, long unixTime) {
		return hasRetFun((client) -> {
			return client.expireAt(key, unixTime);
		}, null);
	}

	public String hget(String key, String field) {
		return hasRetFun((client) -> {
			return client.hget(key, field);
		}, "");
	}

	public Long hset(String key, String field, String value) {
		return hasRetFun((client) -> {
			return client.hset(key, field, value);
		}, null);
	}

	public boolean hexists(String key, String field) {
		return hasRetFun((client) -> {
			return client.hexists(key, field);
		}, false);
	}

	public boolean hdel(String key, String field) {
		return hasRetFun((client) -> {
			return client.hdel(key, field) == 1;
		}, false);
	}

	public void hset(String key, Map<String, String> values) {
		noRetFun((client) -> {
			ShardedJedisPipeline p = client.pipelined();
			for (Entry<String, String> entry : values.entrySet()) {
				p.set(entry.getKey(), entry.getValue());
			}
			p.sync();
		});
	}

	public void select(int db) {
		this.db = db;
	}

	private <T> T hasRetFun(Function<ShardedJedis, T> callback, T failedValue) {
		ShardedJedis client = getInstance();
		try {
			return callback.apply(client);
		} catch (Exception e) {
			logger.info("{}", e);
			return failedValue;
		} finally {
			client.close();
		}
	}

	private void noRetFun(Consumer<ShardedJedis> callback) {
		ShardedJedis client = getInstance();
		try {
			callback.notify();
		} catch (Exception e) {
			logger.info("{}", e);
		} finally {
			client.close();
		}
	}
}
